package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/23 09:38
 */
public class Dwd_02_DwdInteractionCommentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdInteractionCommentInfo().init(
            30002,
            2,
            "Dwd_02_DwdInteractionCommentInfo"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 读取 ods_db 数据
        //tEnv.executeSql("ddl和dml 中的增删改");
        //tEnv.sqlQuery("dml 查询语句");
        //1. 读取 ods_db 的数据: 使用 ddl 建表语句
        readOdsDb(tEnv, "Dwd_02_DwdInteractionCommentInfo");
        
        // 2. 过滤出评论表数据
        Table commentInfo = tEnv.sqlQuery(
            "select " +
                " `data`['id'] id," +
                " `data`['user_id'] user_id," +
                " `data`['sku_id'] sku_id," +
                " `data`['appraise'] appraise," +
                " `data`['comment_txt'] comment_txt," +
                " ts," +
                " pt " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='comment_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("comment_info", commentInfo);
        
        // 3. 读取字典表
        readBaseDic(tEnv);
        
        // 4. 维度退化: lookup join
        Table result = tEnv.sqlQuery("select " +
                                        "id, " +
                                        "user_id, " +
                                        "sku_id, " +
                                        "appraise appraise_code, " +
                                        "dic.dic_name appraise_name," +
                                        "comment_txt, " +
                                        "ts " +
                                        "from comment_info ci " +
                                        "join base_dic for system_time as of ci.pt as dic " +
                                        "on ci.appraise=dic.dic_code ");
        // 5. 把结果写出到 kafka 中
        tEnv.executeSql("create table dwd_interaction_comment_info(" +
                            " id string, " +
                            " user_id string, " +
                            " sku_id string, " +
                            " appraise_code string, " +
                            " appraise_name string, " +
                            " comment_txt string, " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INF));
    
        result.executeInsert("dwd_interaction_comment_info");
        
    
    
    }
}
/*
1. 读取 ods_db 的数据: 使用 ddl 建表语句

2. 过滤出评论表数据

3. 把个别字段做维度冗余(维度退化): 从字典表来数据
    appraise

4. 把结果写出到 kafka 中

使用 sql 的方式:
 
 */